package com.xzx.flink.streamapi.sink;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author xinzhixuan
 * @version 1.0
 * @date 2021-08-29 10:41 下午
 */
public class Sink_06_CustomHBase {
    public static final Logger log = LoggerFactory.getLogger(Sink_06_CustomHBase.class);

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        DataStreamSource<String> source = env.fromElements("hello", "world");
        source.print("=====");

        source.addSink(new RichSinkFunction<String>() {

            private Table table;
            private Connection connection = null;

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                org.apache.hadoop.conf.Configuration configuration = HBaseConfiguration.create();
                configuration.set("hbase.zookeeper.quorum", "172.25.129.157,172.25.129.227,172.25.129.155");
                configuration.set("hbase.zookeeper.property.clientPort", "2181");
                connection =  ConnectionFactory.createConnection(configuration);
                table = connection.getTable(TableName.valueOf("xzx_test"));
            }

            @Override
            public void close() throws Exception {
                table.close();
                connection.close();
            }

            @Override
            public void invoke(String value, Context context) throws Exception {
                Put put = new Put(Bytes.toBytes("00" + value));
                log.info("===============data:{}", value);
                put.addColumn(Bytes.toBytes("d"), Bytes.toBytes("k"), Bytes.toBytes(value));
                table.put(put);
            }
        });

        env.execute(Sink_06_CustomHBase.class.getName());
    }
}
